package com.mcc.collect.opc.service.impl;


import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import com.mcc.collect.cache.domain.PlcCache;
import com.mcc.collect.cache.service.IPlcCacheService;
import com.mcc.collect.config.GlobalVar;
import com.mcc.collect.opc.bootstrap.OpcUAClientRunner;
import com.mcc.collect.opc.service.IHandleService;
import com.mcc.collect.opc.service.IOpcUAClientService;
import com.mcc.collect.opc.utils.TimingUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.SessionActivityListener;
import org.eclipse.milo.opcua.sdk.client.api.UaSession;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscriptionManager;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.*;
import org.eclipse.milo.opcua.stack.core.types.enumerated.*;
import org.eclipse.milo.opcua.stack.core.types.structured.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Service;

import javax.jms.JMSException;
import javax.jms.TextMessage;
import java.util.*;
import java.util.concurrent.ExecutionException;

import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;

@Service
@Slf4j
public class OpcUAClientServiceImpl implements IOpcUAClientService {

    @Autowired
    private IHandleService handleService;

    @Autowired
    private IPlcCacheService plcCacheService;

    @Autowired
    private OpcUAClientRunner opcUAClientRunner;

    @Override
    public void run(OpcUaClient client) throws Exception {
        client.connect().get();
        listenSession(client);
        log.info("开始订阅");
        readNode(client);
        listenSubscription(client);
        GlobalVar.initOpc = true;
    }

    @Override
    public void readConfig() {
        //读取配置文件的变量清单
        List<PlcCache> cacheList = plcCacheService.getOpcReadCacheList();

        long t1 = System.currentTimeMillis();

        Map<String, Object> valueMap = new HashMap<>();
        for (int i = 0; i < cacheList.size(); i++) {
            PlcCache plcCache = cacheList.get(i);
            GlobalVar.itemMap.put(plcCache.getCacheAddr(), plcCache);
            GlobalVar.codeMap.put(plcCache.getCacheCode(), plcCache.getCacheAddr());
            if (StrUtil.isNotEmpty(plcCache.getCacheValue())) {
                Object cacheValue = plcCache.getCacheValue();
                if (plcCache.getCacheType() == 1) {
                    cacheValue = Convert.toDouble(cacheValue, 0D);
                }
                if (plcCache.getCacheType() == 2) {
                    cacheValue = Convert.toInt(cacheValue, 0);
                }
                if (plcCache.getCacheType() == 3) {
                    cacheValue = Convert.toStr(cacheValue, "");
                }
                if (plcCache.getCacheType() == 4) {
                    cacheValue = Convert.toStr(cacheValue, "");
                }
                valueMap.put("plc_cache:" + plcCache.getCacheCode(), cacheValue);
            }
        }
        if (valueMap.size() > 0)
            plcCacheService.updateRedisCache(valueMap);

        TimingUtil.record(t1, "初始化OPC变量");

        GlobalVar.initOpc = true;
    }

    private void listenSession(OpcUaClient client) {
        //监听连接异常
        client.addSessionActivityListener(new SessionActivityListener() {
            @Override
            public void onSessionActive(UaSession session) {
                log.info("Connecting OPC");
            }

            @Override
            public void onSessionInactive(UaSession session) {
                log.info("Disconnecting OPC");
            }
        });
    }

    @Override
    public void readNode(OpcUaClient client) {
        try {
            client.getSubscriptionManager().clearSubscriptions();
            //创建发布间隔100ms的订阅对象
            UaSubscription subscription = client.getSubscriptionManager().createSubscription(100.0).get();

            //创建监控项请求
            List<MonitoredItemCreateRequest> requests = new ArrayList<>();

            //该请求最后用于创建订阅。
            int index = 0;
            for (Map.Entry<String, PlcCache> entry : GlobalVar.itemMap.entrySet()) {
                PlcCache plcCache = entry.getValue();
                if (plcCache != null) {
                    if (Convert.toInt(plcCache.getCacheWay(), 0) != 1) continue;
                    if (StrUtil.isBlank(plcCache.getCacheAddr())) continue;
                }

                index++;
                //创建监控的参数
                MonitoringParameters parameters = new MonitoringParameters(
                        uint(index),
                        // 发布间隔
                        100.0,
                        // filter, 空表示用默认值
                        null,
                        // 队列大小
                        uint(10),
                        //放弃旧配置
                        true
                );

                //创建订阅的变量
                NodeId nodeId = new NodeId(2, entry.getKey());
                ReadValueId readValueId = new ReadValueId(nodeId, AttributeId.Value.uid(), null, null);
                MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting, parameters);
                requests.add(request);
            }

            //创建监控项，并且注册变量值改变时候的回调函数。
            List<UaMonitoredItem> monitoredItems = subscription.createMonitoredItems(
                    TimestampsToReturn.Both,
                    requests,
                    (item, id) -> item.setValueConsumer(this::onSubscriptionValue)
            ).get();

            for (UaMonitoredItem monitoredItem : monitoredItems) {
                if (monitoredItem.getStatusCode().isGood()) {
//                    log.info("订阅成功 nodeId={}", monitoredItem.getReadValueId().getNodeId());
                } else {
                    log.warn(
                            "订阅失败 nodeId={} (status={})",
                            monitoredItem.getReadValueId().getNodeId(), monitoredItem.getStatusCode());
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    private void listenSubscription(OpcUaClient client) {
        client.getSubscriptionManager().addSubscriptionListener(new UaSubscriptionManager.SubscriptionListener() {
            @Override
            public void onSubscriptionTransferFailed(UaSubscription subscription, StatusCode statusCode) {
                log.info("重新订阅");
                try {
                    OpcUaClient client = opcUAClientRunner.getClient();
                    readNode(client);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private void onSubscriptionValue(UaMonitoredItem item, DataValue value) {
        String nodeId = item.getReadValueId().getNodeId().getIdentifier().toString();
        Object nodeValue = value.getValue().getValue();
        updateValue(nodeId, nodeValue);
    }

    public void updateValue(String nodeId, Object nodeValue) {
        nodeValue = nodeValue != null ? nodeValue : null;
        PlcCache plcCache = GlobalVar.itemMap.get(nodeId);
        String code = plcCache.getCacheCode();
        updateValue(code, plcCache, nodeValue);
    }

    @Override
    public void updateValue(String code, PlcCache plcCache, Object nodeValue) {
        //获取前一个值
        Object preValue = plcCacheService.getRedisCache(code);

        //类型转换
        if (nodeValue != null) {
            if (StrUtil.equals(nodeValue.toString(), "true")) nodeValue = 1;
            if (StrUtil.equals(nodeValue.toString(), "false")) nodeValue = 0;
        }
        if (plcCache.getCacheType() == 1) {
            nodeValue = NumberUtil.round(Convert.toDouble(nodeValue, 0D), 3).doubleValue();
        }
        if (plcCache.getCacheType() == 2) {
            nodeValue = Convert.toInt(nodeValue, 0);
        }
        if (plcCache.getCacheType() == 3) {
            nodeValue = Convert.toStr(nodeValue, "");
        }
        if (plcCache.getCacheType() == 4) {
            //日期
            if (nodeValue != null) {
                long time = ((DateTime) nodeValue).getJavaTime();
                nodeValue = DateUtil.formatDateTime(new Date(time));
            } else {
                nodeValue = "";
            }
        }
        //更新redis
        plcCacheService.updateRedisCache(code, nodeValue);

        Date dt = new Date();
        //状态变量保存
        handleService.saveNode(code, nodeValue, dt, plcCache, preValue);
        //mq异步推送
        handleService.pushMq(code, nodeValue, dt, plcCache.getCacheType(), preValue);
    }

    //写入变量
    @Override
    public boolean writeValue(OpcUaClient client, String code, Object value) {
        if (!GlobalVar.initOpc) return false;
        if (client == null) return false;
        String nodeIdStr = GlobalVar.codeMap.get(code);
        if (StrUtil.isEmpty(nodeIdStr)) return false;
        NodeId nodeId = new NodeId(2, nodeIdStr);
        DataValue dv = new DataValue(new Variant(value), null, null);
        try {
            StatusCode statusCode = client.writeValue(nodeId, dv).get();
            boolean res = statusCode.isGood();
            if (!res)
                log.info("{}：{}", nodeIdStr, statusCode);
            return res;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return false;
    }

    //监听推送数据，调用跟踪逻辑
    @JmsListener(destination = "DATA.TRACK", containerFactory = "jmsQueue")
    public void listenData(TextMessage message) {
        try {
            JSONObject json = new JSONObject(message.getText());
            String code = json.getStr("code");
            Object nodeValue = json.get("nodeValue");
            Date dt = json.getDate("dt");
            Integer cacheType = json.getInt("cacheType");
            Object preValue = json.get("preValue");
            handleService.oneUpdateTrigger(code, nodeValue, dt, cacheType, preValue);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
